package com.allen.flink.batch;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 功能: 广播变量
 *
 * @date: 2020-03-24 14:16
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class Broadcast {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //准备需要广播的数据
        ArrayList<Tuple2<String, String>> list = new ArrayList<>();
        list.add(new Tuple2<>("101", "Mars"));
        list.add(new Tuple2<>("102", "Tom"));
        list.add(new Tuple2<>("103", "Alan"));
        
        //读取数据源
        DataSet<Tuple2<String, String>> broadCastData = env.fromCollection(list);
        
        //数据集转换为map
        DataSet<HashMap<String, String>> toBroadCast = broadCastData.map(new MapFunction<Tuple2<String, String>, HashMap<String, String>>() {
            @Override
            public HashMap<String, String> map(final Tuple2<String, String> value) throws Exception {
                
                HashMap<String, String> out = new HashMap<>();
                out.put(value.f0, value.f1);
                return out;
            }
        });
        
        //准备处理数据  金币数量
        ArrayList<Tuple2<String, Integer>> operatorList = new ArrayList<>();
        operatorList.add(new Tuple2<>("101", 2000));
        operatorList.add(new Tuple2<>("102", 1500));
        operatorList.add(new Tuple2<>("103", 1000));
        
        //读取处理数据
        DataSet<Tuple2<String, Integer>> operatorData = env.fromCollection(operatorList);
        
        //转换为map
        DataSet<String> result = operatorData.map(new RichMapFunction<Tuple2<String, Integer>, String>() {
            List<HashMap<String, String>> broadCastMap = new ArrayList<>();
            
            Map<String, String> allMap = new HashMap<>();
            
            @Override
            public String map(final Tuple2<String, Integer> value) throws Exception {
                String userName = allMap.get(value.f0);
                return userName + "," + value.f1;
            }
            
            @Override
            public void open(final Configuration parameters) throws Exception {
                super.open(parameters);
                //获取广播数据
                this.broadCastMap = getRuntimeContext().getBroadcastVariable("userInfo");
                
                for (HashMap map : broadCastMap) {
                    allMap.putAll(map);
                }
                
            }
        }).withBroadcastSet(toBroadCast, "userInfo");
        
        result.print();
    }
    
}
